package com.young.springboot.kafka;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.util.unit.DataSize;
import org.springframework.util.unit.DataUnit;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

/**
 * @description：
 * @author: yangyahui01
 * @date: 12/5/24 2:37 PM
 */
public class KafkaClientProperties extends KafkaProperties {

    private final Consumer consumer = new Consumer();

    private final Producer producer = new Producer();

    private final Listener listener = new Listener();

    private final Admin admin = new Admin();

    /**
     * 控制客户端等待请求响应的最长时间，如果在指定的超时时间内没有收到响应，客户端将采取以下两种行动之一
     * 1）重新发送请求：如果重试次数尚未耗尽，客户端将重新发送请求
     * 2）请求失败：如果重试次数已经耗尽，客户端将使请求失败
     */
    private Duration requestTimeout = Duration.ofMillis(60000);

    /**
     * 指定在多久之后关闭空闲的连接，默认 9 分钟
     */
    private Duration connectionsMaxIdle = Duration.ofMillis(540000);

    /**
     * 表示在尝试重新连接到一个连续连接失败的代理时，等待的最大时间（以毫秒为单位）
     */
    private Duration reconnectBackoff = Duration.ofMillis(500);

    /**
     * 表示尝试重新连接到给定主机之前等待的基本时间量
     */
    private Duration reconnectBackoffMax = Duration.ofMillis(3000);

    /**
     * 指定在重试失败的请求之前等待的时间量（以毫秒为单位）
     */
    private Duration retryBackoff = Duration.ofMillis(100);

    private boolean enableTransaction = true;

    public Duration getRequestTimeout() {
        return requestTimeout;
    }

    public void setRequestTimeout(Duration requestTimeout) {
        this.requestTimeout = requestTimeout;
    }

    public Duration getConnectionsMaxIdle() {
        return connectionsMaxIdle;
    }

    public void setConnectionsMaxIdle(Duration connectionsMaxIdle) {
        this.connectionsMaxIdle = connectionsMaxIdle;
    }

    public Duration getReconnectBackoff() {
        return reconnectBackoff;
    }

    public void setReconnectBackoff(Duration reconnectBackoff) {
        this.reconnectBackoff = reconnectBackoff;
    }

    public Duration getReconnectBackoffMax() {
        return reconnectBackoffMax;
    }

    public void setReconnectBackoffMax(Duration reconnectBackoffMax) {
        this.reconnectBackoffMax = reconnectBackoffMax;
    }

    public Duration getRetryBackoff() {
        return retryBackoff;
    }

    public void setRetryBackoff(Duration retryBackoff) {
        this.retryBackoff = retryBackoff;
    }

    public boolean isEnableTransaction() {
        return enableTransaction;
    }

    public void setEnableTransaction(boolean enableTransaction) {
        this.enableTransaction = enableTransaction;
    }

    public Consumer getConsumer() {
        return consumer;
    }

    public Producer getProducer() {
        return producer;
    }

    public Listener getListener() {
        return listener;
    }


    protected Map<String, Object> buildCommonProperties() {
        Map<String, Object> properties = new HashMap<>();
        properties.putAll(super.buildCommonProperties());
        properties.put(CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG,(int)this.requestTimeout.toMillis());
        properties.put(CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG,(int)this.connectionsMaxIdle.toMillis());
        properties.put(CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_CONFIG,(int)this.reconnectBackoffMax.toMillis());
        properties.put(CommonClientConfigs.RECONNECT_BACKOFF_MS_CONFIG,(int)this.reconnectBackoff.toMillis());
        properties.put(CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG,(int)this.retryBackoff.toMillis());
        return properties;
    }

    public Map<String, Object> buildProducerProperties() {
        Map<String, Object> properties = buildCommonProperties();
        properties.putAll(this.producer.buildProperties());
        return properties;
    }

    public Map<String, Object> buildAdminProperties() {
        Map<String, Object> properties = buildCommonProperties();
        properties.putAll(this.admin.buildProperties());
        return properties;
    }


    public Map<String, Object> buildConsumerProperties() {
        Map<String,Object> properties = buildCommonProperties();
        properties.putAll(this.consumer.buildProperties());
        return properties;
    }

    public static class Consumer extends KafkaProperties.Consumer {

        private int maxPollRecords = 100;

        private Duration fetchMaxWait = Duration.ofMillis(30000);

        private Duration maxPollInterval = Duration.ofMillis(120000);//2分钟

        private Duration sessionTimeout = Duration.ofMillis(30000);

        private String autoOffsetReset = "latest";

        private boolean enableAutoCommit = false;

        @Override
        public Integer getMaxPollRecords() {
            return maxPollRecords;
        }

        public void setMaxPollRecords(int maxPollRecords) {
            this.maxPollRecords = maxPollRecords;
        }

        public Duration getMaxPollInterval() {
            return maxPollInterval;
        }

        public void setMaxPollInterval(Duration maxPollInterval) {
            this.maxPollInterval = maxPollInterval;
        }

        public Duration getSessionTimeout() {
            return sessionTimeout;
        }

        public void setSessionTimeout(Duration sessionTimeout) {
            this.sessionTimeout = sessionTimeout;
        }

        @Override
        public String getAutoOffsetReset() {
            return autoOffsetReset;
        }

        @Override
        public void setAutoOffsetReset(String autoOffsetReset) {
            this.autoOffsetReset = autoOffsetReset;
        }

        public boolean isEnableAutoCommit() {
            return enableAutoCommit;
        }

        public void setEnableAutoCommit(boolean enableAutoCommit) {
            this.enableAutoCommit = enableAutoCommit;
        }

        @Override
        public Duration getFetchMaxWait() {
            return fetchMaxWait;
        }

        @Override
        public void setFetchMaxWait(Duration fetchMaxWait) {
            this.fetchMaxWait = fetchMaxWait;
        }

        public Map<String, Object> buildProperties() {
            Map<String,Object> properties = super.buildProperties();
            properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,this.maxPollRecords);
            properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,(int)this.maxPollInterval.toMillis());
            properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG,(int)this.sessionTimeout.toMillis());
            properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,this.autoOffsetReset);
            properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,this.enableAutoCommit);
            properties.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG,(int)this.fetchMaxWait.toMillis());
            return properties;
        }

    }

    public static class Producer extends KafkaProperties.Producer {

        private String acks = "all";

        /**
         * 表示生产者在等待缓冲区可用或等待元数据的过程中能阻塞的最长时间，这里为 30s
         */
        private int maxBlock = 30000;
        private int retries = 3;//最大重试次数
        private DataSize batchSize = DataSize.of(8192, DataUnit.BYTES);//8K
        private int linger = 500;
        /**
         * 生产者发送给 Kafka 服务器请求的最大大小（以字节为单位）。
         */
        private int maxRequestSize = 1048576;
        private int transactionTimeout = 30000;

        @Override
        public String getAcks() {
            return acks;
        }

        @Override
        public void setAcks(String acks) {
            this.acks = acks;
        }

        public int getMaxBlock() {
            return maxBlock;
        }

        public void setMaxBlock(int maxBlock) {
            this.maxBlock = maxBlock;
        }

        @Override
        public Integer getRetries() {
            return retries;
        }

        public void setRetries(int retries) {
            this.retries = retries;
        }

        @Override
        public DataSize getBatchSize() {
            return batchSize;
        }

        public void setBatchSize(DataSize batchSize) {
            this.batchSize = batchSize;
        }

        public int getLinger() {
            return linger;
        }

        public void setLinger(int linger) {
            this.linger = linger;
        }

        public int getMaxRequestSize() {
            return maxRequestSize;
        }

        public void setMaxRequestSize(int maxRequestSize) {
            this.maxRequestSize = maxRequestSize;
        }

        public int getTransactionTimeout() {
            return transactionTimeout;
        }

        public void setTransactionTimeout(int transactionTimeout) {
            this.transactionTimeout = transactionTimeout;
        }

        public Map<String, Object> buildProperties() {
            Map<String,Object> properties = super.buildProperties();
            properties.put(ProducerConfig.ACKS_CONFIG,this.acks);
            properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG,this.maxBlock);
            properties.put(ProducerConfig.BATCH_SIZE_CONFIG,(int)this.batchSize.toBytes());
            properties.put(ProducerConfig.LINGER_MS_CONFIG,this.linger);
            properties.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG,this.maxRequestSize);
            properties.put(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,this.transactionTimeout);
            return properties;
        }

    }

    public class Listener extends KafkaProperties.Listener {


        private Integer concurrency = 2;

        private Duration pollTimeout = Duration.ofMillis(200);

        private boolean batchListener = true;

        private boolean enableAutoRestart = true;

        @Override
        public Integer getConcurrency() {
            return concurrency;
        }

        @Override
        public void setConcurrency(Integer concurrency) {
            this.concurrency = concurrency;
        }

        @Override
        public Duration getPollTimeout() {
            return pollTimeout;
        }

        public void setPollTimeout(Duration pollTimeout) {
            this.pollTimeout = pollTimeout;
        }

        public boolean isBatchListener() {
            return batchListener;
        }

        public void setBatchListener(boolean batchListener) {
            this.batchListener = batchListener;
        }

        public boolean isEnableAutoRestart() {
            return enableAutoRestart;
        }

        public void setEnableAutoRestart(boolean enableAutoRestart) {
            this.enableAutoRestart = enableAutoRestart;
        }
    }

}
